/**
 * 
 */
package cn.g31010.sip.util;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.SocketException;
import java.util.concurrent.LinkedBlockingQueue;

import com.gqf.common.net.UDPSocket;
import com.gqf.common.util.Converts;
import java.util.concurrent.TimeUnit;

/**
 * @author user
 * 
 */
public class MessageQueue<E> implements Runnable {

	private static final int DEFAULT_MAX_LENGTH = 1000;
	private int maxLength = DEFAULT_MAX_LENGTH;
	private LinkedBlockingQueue<E> queue = new LinkedBlockingQueue<E>(maxLength);
	private UDPSocket udp;
	private boolean listen;

	public MessageQueue() throws SocketException {
		udp = new UDPSocket();
	}

	public MessageQueue(int port) throws SocketException {
		udp = new UDPSocket(port);
	}

	public synchronized void listen() {
		if(!listen){
			Thread daemon = new Thread(this);
			daemon.setDaemon(true);
			daemon.start();
			listen=true;
		}
	}

	public void send(String address, int port, Object obj) throws IOException {
		byte[] b = Converts.objectToBytes((java.io.Serializable)obj);
		DatagramPacket packet = UDPSocket.createPacket(b, address, port);
		udp.send(packet);
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.lang.Runnable#run()
	 */
	@SuppressWarnings("unchecked")
	public void run() {
		while (true) {
			try {
				DatagramPacket packet = udp.receive(0);
				int len = packet.getLength();
				byte[] b = packet.getData();
				if (b.length > len) {
					byte[] bb = new byte[len];
					System.arraycopy(b, 0, bb, 0, len);
					b = bb;
				}
				Object obj = Converts.bytesToObject(b);
				boolean success = queue.offer((E) obj);
				if(!success){
					System.err.println("ERROR: queue is overflow.");
				}
			} catch (IOException e) {
				e.printStackTrace();
			} catch (ClassNotFoundException e) {
				
			}
		}
	}

	public int getMaxLength() {
		return this.maxLength;
	}

	public void setMaxLength(int maxLength) {
		this.maxLength = maxLength;
	}
	
	public Object receive(int milliseconds){
		try {
			return queue.poll(milliseconds, TimeUnit.MILLISECONDS);
		} catch (InterruptedException e) {
			return null;
		}
	}

}
